package march.firstmarch21.march02;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;
import java.util.Date;

public class CallMapper extends Mapper<LongWritable, Text,Text,CallBean> {
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

        //1.将一条数据分为打电话和接电话两个类
        CallBean callBean = new CallBean();
        CallBean acceptBean = new CallBean();
        //2.取出相应的数据
        String line = value.toString();
        String[] split = line.split(",");
        String callPhone = split[0];
        String acceptPhone = split[1];
        //3.处理时间长度
        Date time1 = new Date(Long.parseLong(split[2]) * 1000L);
        Date time2 = new Date(Long.parseLong(split[3]) * 1000L);
        long time = (time2.getTime() - time1.getTime()) / 1000;
        //4.填充对象
        callBean.setPhone(callPhone);
        callBean.setCallLen(Long.toString(time));
        callBean.setFlag("call");
        acceptBean.setPhone(acceptPhone);
        acceptBean.setAcceptLen(Long.toString(time));
        acceptBean.setFlag("accept");
        //5.将两个对象分别输出（这里是重点）
        context.write(new Text(callPhone), callBean);
        context.write(new Text(acceptPhone), acceptBean);
    }
}





